﻿using System.Threading.Channels;

namespace Digitalmes.Infrastructure.Common.Channels;

/// <summary>
/// Broker 基类
/// </summary>
/// <typeparam name="T"></typeparam>
public abstract class BrokerBase<T>
{
    private readonly Channel<T> _queue;

    /// <summary>
    /// 初始化一个新的 Channel。
    /// </summary>
    /// <param name="capacity">容量</param>
    /// <param name="mode">通道满载后的处理模式</param>
    /// <param name="isSingleReader">是否只允许单一对象读取</param>
    public BrokerBase(int capacity, BrokerMode mode = BrokerMode.Wait, bool isSingleReader = true)
    {
        // 多个写入端，单个读取端
        _queue = Channel.CreateBounded<T>(new BoundedChannelOptions(capacity)
        {
            FullMode = (BoundedChannelFullMode)(int)mode,
            SingleReader = isSingleReader,
            SingleWriter = false,
        });
    }

    /// <summary>
    /// 将事件写入队列。
    /// </summary>
    /// <param name="item">加入队列的事件项</param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async ValueTask QueueAsync(T item, CancellationToken cancellationToken = default)
    {
        await _queue.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>
    /// 异步读取，若队列中没有数据项时会异步阻塞当前线程。
    /// </summary>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async ValueTask<T> DequeueAsync(CancellationToken cancellationToken = default)
    {
        // 效果等于 WaitToReadAsync + TryRead
        return await _queue.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
    }
}
